package com.bjsdzk.collector.one2one;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.DelayQueue;
import java.util.function.Predicate;
import java.util.function.Supplier;

import com.alibaba.dubbo.config.annotation.Reference;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.bjsdzk.backend.rpc.before.OneNETbefore;
import com.bjsdzk.backend.rpc.serial.SeroalportPlatformService;
import com.bjsdzk.collector.config.MqConfig;
import com.bjsdzk.collector.constant.Constant;
import com.bjsdzk.collector.entity.DelayedList;
import com.bjsdzk.collector.serviceImp.DictionariesService;
import com.bjsdzk.collector.serviceImp.ExceColleContrService;
import com.bjsdzk.collector.serviceImp.ParamsSettingService;
import com.bjsdzk.collector.utils.PublicUtils;
import io.vertx.core.Verticle;
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonArray;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.Order;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import io.vertx.core.DeploymentOptions;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.redis.RedisClient;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * @author pc
 */
@Component
@Order(2)
public class StartModServer implements CommandLineRunner {
    @Autowired
    private Vertx vertx;

    @Autowired
    private RedisClient redis;
    @Autowired
    private MongoClient mongoClient;
    @Reference(check = false, version = "1.0.0",group = Constant.GROUP_PLATFORM, lazy = true ,retries = -1)
    private SeroalportPlatformService seroalportPlatformService;
    @Reference(check = false, version = "1.0.0",group = Constant.GROUP_APP, lazy = true ,retries = -1)
    private SeroalportPlatformService aPPseroalportPlatformService;
    @Reference(check = false, version = "1.0.0",group = Constant.GROUP_BANK, lazy = true ,retries = -1)
    private SeroalportPlatformService bankseroalportPlatformService;
    @Reference(check = false, version = "1.0.0",group = Constant.GROUP_MINIAPP_ELECTRICENGINEER, lazy = true ,retries = -1)
    private SeroalportPlatformService miniappSeroalportPlatformService;
    @Reference( check = false,version ="1.0.0",lazy=true,retries = 1,group = Constant.GROUP_PLATFORM)
    private OneNETbefore oneNETbefore;

    @Autowired
    private ExceColleContrService exceColleContrService;
    @Autowired
    private ParamsSettingService paramsSettingService;
    @Autowired
    private DictionariesService dictionariesService;
    @Autowired
    ApplicationContext applicationContext;
    @Value("${rpc.switch}")
    private boolean rpcSwitch;
    @Autowired
    private ProducerBean producer;
    @Autowired
    private MqConfig mqConfig;
    @Autowired
    private Environment env;

    private List<SeroalportPlatformService> seroalportPlatformServices=new ArrayList<>();

    private static Logger logger = LoggerFactory.getLogger(StartModServer.class);

    @PostConstruct
    public void postConstruct(){
        seroalportPlatformServices.add(seroalportPlatformService);
        seroalportPlatformServices.add(bankseroalportPlatformService);
        seroalportPlatformServices.add(aPPseroalportPlatformService);
        seroalportPlatformServices.add(miniappSeroalportPlatformService);
    }

    @Override
    public void run(String... args) throws Exception {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        DelayQueue<DelayedList> queue = new DelayQueue<>();
        Map<String, Integer> startFail = new HashMap<>(16);
        long lon = 50;

        redis.keys("mod_*", res -> {
            if (res.succeeded()) {
                JsonArray result = res.result();
                logger.info("mod start count :{}", result.size());
                long l = 0;
                for (int i = 0; i < result.size(); i++) {
                    queue.offer(new DelayedList(i, result.getString(i), l = lon + l));
                }
                countDownLatch.countDown();
            }
        });

        countDownLatch.await();

        while (true) {
            if (queue.isEmpty()) {
                logger.info("mod all start finsh ");
                break;
            }
            DelayedList ele = queue.poll();
            if (ele != null) {
                redis.hgetall(ele.getBody(), res -> {
                    startModServer(res.result());
                });
            }

        }

        EventBus eventBus = vertx.eventBus();

        eventBus.consumer(Constant.MOD_START_FAIL, msg -> {
            JsonObject body = (JsonObject) msg.body();
            logger.debug("MOD_START_FAIL:" + body);
            String port = body.getString("port");
            Integer count = startFail.get(port);
            startFail.put(port, count = count == null ? 1 : ++count);
            if (count <= 10) {
                vertx.setTimer(count * 10 * 1000, r -> startModServer(body));
            } else {
                logger.error("deployment fail three times ==> port:" + port + " ==> ");
            }
        });

        eventBus.consumer(Constant.MOD_ONLINE_STATUS, msg -> {
            invokeRpc(msg, "MOD_ONLINE_STATUS:", (str) -> {
                SeroalportPlatformService serviceBySource = getServiceBySource((JsonObject) msg.body(), seroalportPlatformServices);
                return invokeRpcBySource(() -> serviceBySource.statusChange(str));
            });
        });
        eventBus.consumer(Constant.MOD_OFFLINE_STATUS, msg -> {
            invokeRpc(msg, "MOD_OFFLINE_STATUS:", (str) -> {
                SeroalportPlatformService serviceBySource = getServiceBySource((JsonObject) msg.body(), seroalportPlatformServices);
                return invokeRpcBySource(() -> serviceBySource.statusChange(str));
            });
        });

        eventBus.consumer(Constant.MOD_LATEST, msg -> {
            invokeRpc(msg, "MOD_LATEST:", (str) -> {
                SeroalportPlatformService serviceBySource = getServiceBySource((JsonObject) msg.body(), seroalportPlatformServices);
                return invokeRpcBySource(() -> serviceBySource.data(str));
            });
        });

        eventBus.consumer(Constant.MOD_DEVICE485_FAIL, msg -> {
            invokeRpc(msg, "MOD_DEVICE485_FAIL:", (str) -> {
                SeroalportPlatformService serviceBySource = getServiceBySource((JsonObject) msg.body(), seroalportPlatformServices);
                return invokeRpcBySource(() -> serviceBySource.faultChannel(str));
            });
        });

        eventBus.consumer(Constant.MOD_DEVICE_FAIL, msg -> {
            invokeRpc(msg, "MOD_DEVICE_FAIL:", str -> {
                JsonObject data = (JsonObject) msg.body();
                SeroalportPlatformService serviceBySource = getServiceBySource(data, seroalportPlatformServices);
                return invokeRpcBySource(() -> serviceBySource.fault(str));
            });
        });
        eventBus.consumer(Constant.MOD_DEVICE_POLICE, msg -> {
            invokeRpc(msg, "MOD_DEVICE_POLICE:", (str) -> {
                JsonObject data = (JsonObject) msg.body();
                SeroalportPlatformService serviceBySource = getServiceBySource(data, seroalportPlatformServices);
                return invokeRpcBySource(() -> serviceBySource.police(str));
            });
        });

        eventBus.consumer(Constant.MOD_DEVICE_CLEANPOLICE, msg -> {
            invokeRpc(msg, "MOD_DEVICE_CLEANPOLICE:", (str) -> {
                JsonObject data = (JsonObject) msg.body();
                SeroalportPlatformService serviceBySource = getServiceBySource(data, seroalportPlatformServices);
                return invokeRpcBySource(() -> serviceBySource.cleanPolice(str));
            });
        });

        eventBus.consumer(Constant.MOD_CALLBACK, msg -> {
            invokeRpc(msg, "MOD_CALLBACK:", (str) -> {
                JsonObject body = (JsonObject) msg.body();
                JsonObject data = null;
                try {
                    data = body.getJsonObject("data");
                } catch (Exception e) {

                }
                SeroalportPlatformService serviceBySource = getServiceBySource(data, seroalportPlatformServices);
                return invokeRpcBySource( () -> serviceBySource.callback(str));

            });
        });
    }

    protected SeroalportPlatformService getServiceBySource(JsonObject data,List<SeroalportPlatformService> seroalportPlatformServices) {
        String source = data.getString("source");
        SeroalportPlatformService seroalportPlatformService ;
        if(data==null|| StringUtils.equals(Constant.GROUP_PLATFORM,source)){
            seroalportPlatformService=seroalportPlatformServices.get(0);
        }else if(StringUtils.equals(Constant.GROUP_BANK, source)){
            seroalportPlatformService=seroalportPlatformServices.get(1);
        }else if(StringUtils.equals(Constant.GROUP_APP, source)) {
            seroalportPlatformService=seroalportPlatformServices.get(2);
        }else if(StringUtils.equals(Constant.GROUP_MINIAPP_ELECTRICENGINEER, source)) {
            seroalportPlatformService=seroalportPlatformServices.get(3);
        }else {
            seroalportPlatformService=seroalportPlatformServices.get(0);
        }
        return seroalportPlatformService;
    }

    protected boolean invokeRpcBySource(Supplier<Boolean> platformService) {
        boolean result = platformService.get();
        return result;
    }

    public void invokeRpc(Message<Object> msg, String opertion, Predicate<String> supplier) {
        JsonObject body = (JsonObject) msg.body();
        int count = 1;
        checkError(count, (c) -> {
            logger.info( "{}==>send times:{}==>params:{}",opertion,c,body);
            if(rpcSwitch) {
                supplier.test(body.toString());
            }
             return true;
        });
    }

    public void checkError(int count, Predicate<Integer> predicate) {
        boolean result = false;
        try {
            predicate.test(count);
            result = true;
        } catch (Exception e) {
            PublicUtils.printException(e, -1);
        }
        count++;
        int c = count;
        if (!result && count < 1) {
            vertx.setTimer(count * 1000 * 60, h -> {
                checkError(c, predicate);
            });

        }
    }

    public void startModServer(JsonObject jsonObject) {

        String port = jsonObject.getString("port");
        String dtuType = jsonObject.getString("dtuType");
        Verticle verticle = null;
//        Map<String, BasetVerticle> beansOfType = applicationContext.getBeansOfType(BasetVerticle.class);
        switch (dtuType) {
            case Constant.DTUTYPE_6:
            case Constant.DTUTYPE_8:
                verticle = new ModbusModuleServer(redis, mongoClient);
                break;
            case Constant.DTUTYPE_7:
                verticle = new OnlyServer(redis, mongoClient);
                break;
            case Constant.DTUTYPE_10:
                verticle = new BasetVerticleServer(redis, mongoClient, exceColleContrService, paramsSettingService, dictionariesService);
                break;
            case Constant.DTUTYPE_11:
            case Constant.DTUTYPE_12:
                verticle = new LnServer(redis, mongoClient, exceColleContrService, paramsSettingService, dictionariesService);
                break;
            case Constant.DTUTYPE_13:
                verticle = new MandunVerticle(redis, mongoClient, exceColleContrService, paramsSettingService, dictionariesService,producer,mqConfig,env);
                break;
            case Constant.DTUTYPE_14:
                verticle = new ZhuowenYanshiVerticleServer(redis, mongoClient, exceColleContrService, paramsSettingService, dictionariesService,oneNETbefore);
                break;

        }


        DeploymentOptions deploymentOptions = new DeploymentOptions().setWorkerPoolSize(80)
                .setConfig(jsonObject);
        vertx.deployVerticle(verticle, deploymentOptions, id -> {
            if (id.succeeded()) {
                logger.info("Verticle: ModbusModuleServer" + "==>port:{}==>" + id.result() + " deployed!", port);
            } else {
                logger.error("Verticle: ModbusModuleServer" + "==>port:{}==>" + id.cause().getMessage() + " deployed!", port);
            }

        });
    }

    @PreDestroy
    public void preDestroy() {
        redis.set("qidong", "0", h ->{
        });
        logger.info("stop project--------------------------=================");
    }
    @PostConstruct
    public void init() {
        redis.set("qidong", "1", h ->{
        });
        logger.info("start project--------------------------=================");
    }

}

